package com.group1.realtime.util;

import com.alibaba.fastjson.JSONObject;
import com.group1.realtime.annotation.NotSink;
import com.group1.realtime.bean.BroadCastVideoTimesBean;
import com.group1.realtime.bean.TableProcess;
import com.group1.realtime.common.Constant;
import com.group1.realtime.sink.PhoenixSink;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.shaded.guava18.com.google.common.base.CaseFormat;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.Properties;
import java.util.stream.Collectors;

public class FlinkSinkUtil {
    public static SinkFunction<Tuple2<JSONObject, TableProcess>> getPhoenixSink() {

        return new PhoenixSink();
    }

    public static SinkFunction<String> getKafkaSink(String topic) {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", Constant.KAFKA_BROKERS);
        props.put("transaction.timeout.ms", 15 * 60 * 1000);

        return new FlinkKafkaProducer<String>(
                "default",
                new KafkaSerializationSchema<String>() {
                    @Override
                    public ProducerRecord<byte[], byte[]> serialize(String element,
                                                                    @Nullable Long aLong) {
                        return new ProducerRecord<>(topic, element.getBytes(StandardCharsets.UTF_8));
                    }
                },
                props,
                FlinkKafkaProducer.Semantic.NONE
        );
    }

    public static <T> SinkFunction<T> getClickHouseSink(String table,
                                                        Class<T> tClass) {
        String driver = Constant.CLICKHOSUE_DRIVER;
        String url = Constant.CLICKHOSUE_URL;

        // 从class中找到所有的属性名
        Field[] fields = tClass.getDeclaredFields();
        String names = Arrays.stream(fields)
                .filter(field -> {
                    // 把不需要写入到clickhouse的字段提前过滤掉
                    //return  !field.getName().equals("orderIdSet") && !field.getName().equals("sku_id");
                    NotSink noSink = field.getAnnotation(NotSink.class);
                    return noSink == null;

                })
                .map(field -> {
                    // 把java的驼峰缓存下划线
                    return CaseFormat.LOWER_CAMEL.to(CaseFormat.LOWER_UNDERSCORE, field.getName());
                }).collect(Collectors.joining(","));

        StringBuilder sql = new StringBuilder();
        sql
                .append("insert into ")
                .append(table)
                .append("(")
                .append(names)
                .append(")values(")
                .append(names.replaceAll("[^,]+", "?")) // 把字段用?替换
                .append(")");

        System.out.println("clickhouse插入语句: " + sql);

        return getJdbcSink(driver, url, null, null, sql.toString());
    }

    private static <T> SinkFunction<T> getJdbcSink(String driver,
                                                   String url,
                                                   String user,
                                                   String password,
                                                   String sql) {
        return JdbcSink
                .sink(sql,
                        new JdbcStatementBuilder<T>() {
                            @Override
                            public void accept(PreparedStatement ps,
                                               T t) throws SQLException {
                                // 从对象t中取出值, 给 ps(sql)中的占位符进行赋值
                                // TODO  insert into dws_traffic_source_keyword_page_view_window(stt,edt,source,keyword,keyword_count,ts)values(?,?,?,?,?,?)
                                // 根据字段名, 从 t对象中取出对应的值, 给占位符赋值
                                // 获取class: Class.forName  类名.class  对象.getClass
                                Class<?> tClass = t.getClass();
                                Field[] fields = tClass.getDeclaredFields();

                                int position = 1;
                                try {
                                    for (Field field : fields) {
                                        NotSink noSink = field.getAnnotation(NotSink.class);
                                        if (noSink == null) {
                                            field.setAccessible(true);
                                            Object v = field.get(t);
                                            ps.setObject(position++, v);
                                        }
                                    }
                                } catch (IllegalAccessException e) {
                                    e.printStackTrace();
                                }
                            }
                        },
                        new JdbcExecutionOptions.Builder()
                                .withBatchSize(10)
                                .withBatchIntervalMs(2000)
                                .withMaxRetries(3)
                                .build(),
                        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                                .withDriverName(driver)
                                .withUrl(url)
                                .withUsername(user)
                                .withPassword(password)
                                .build()

                );
    }
}

